package wsmsgsubpub

import (
	"bytes"
	"fmt"
	"gitee.com/ymofen/gobase"
	"golang.org/x/net/websocket"
	"sync"
	"sync/atomic"
	"time"
)

type WsBinmsgToken struct {
	token   string
	wait    chan struct{}
	respBuf []byte
}

type WsBinMsg struct {
	Topic string
	Msg   []byte
}

func NewWsBinMsgToken() *WsBinmsgToken {
	rval := &WsBinmsgToken{token: fmt.Sprintf("%s-%d", tokenPrefix, atomic.AddInt32(&tokenSN, 1))}
	rval.wait = make(chan struct{}, 1)
	return rval
}

func (this *WsBinmsgToken) Done() bool {
	select {
	case this.wait <- struct{}{}:
		return true
	default:
		return false
	}
}

func (this *WsBinmsgToken) Wait(t time.Duration) bool {
	select {
	case <-this.wait:
		return true
	case <-time.After(t):
		return false
	}
}

type WsBinMsgCli struct {
	connectDNA   int32
	WaitRespFlag int8
	wscli        *WsClient
	respTimeOut  time.Duration
	cmdLk        sync.RWMutex

	subLk       sync.RWMutex
	subMap      map[string]int32
	cacheCmd    map[string]*WsBinmsgToken
	Msglst      chan *WsBinMsg
	OnConnected func(sender *WsBinMsgCli)
}

func NewWsBinMsgCli(url string) *WsBinMsgCli {
	rval := &WsBinMsgCli{cacheCmd: make(map[string]*WsBinmsgToken), subMap: make(map[string]int32), WaitRespFlag: 1}
	rval.respTimeOut = time.Second * 5
	rval.Msglst = make(chan *WsBinMsg)
	rval.wscli = NewWsClient()
	rval.wscli.OnConnected = rval.onWsCliConnected
	rval.wscli.ConfigWsReqUrl(url, "")
	rval.wscli.OnRecv = rval.onRecv
	return rval
}

func (this *WsBinMsgCli) Open() {
	if this.wscli.CheckStart() {
		time.Sleep(time.Millisecond * 200)
	}
}

func (this *WsBinMsgCli) Close() error {
	this.wscli.Stop()
	this.wscli.WaitFor(time.Second * 5)
	close(this.Msglst)
	return nil
}

func (this *WsBinMsgCli) onWsCliConnected(sender *WsClient) {
	dna := atomic.AddInt32(&this.connectDNA, 1)
	evt := this.OnConnected
	if evt != nil {
		evt(this)
	}
	go this.resubAll(dna)
}

func (this *WsBinMsgCli) RecvMsg() *WsBinMsg {
	return <-this.Msglst
}

func (this *WsBinMsgCli) onRecv(cli *WsClient, msgtype byte, buf []byte) {
	r := bytes.NewReader(buf)
	s0, err := DecodeStrFromMsgPack(r)
	if err != nil {
		cli.RequestDisconnect("非法的数据包")
		return
	}
	head := gobase.NewStrMap()
	head.URLFormDecode(s0)
	token := head.StringByName("token", "")

	bodyBuf, err := DecodeBytesFromMsgPack(r)
	if err != nil {
		cli.RequestDisconnect("非法的数据包")
		return
	}

	if len(token) > 0 {
		this.cmdLk.RLock()
		cmd := this.cacheCmd[token]
		this.cmdLk.RUnlock()
		if cmd != nil {
			cmd.respBuf = bodyBuf
			cmd.Done()
		}
	} else {
		msg := &WsBinMsg{Topic: head.StringByName("topic", ""), Msg: bodyBuf}
		this.Msglst <- msg
	}
}

func (this *WsBinMsgCli) innerSub(topic string) error {
	token := NewWsBinMsgToken()
	var cache gobase.BytesBuilder
	cmdMap := gobase.NewStrMap()
	cmdMap["token"] = token.token
	cmdMap["topics"] = topic
	cmdMap["action"] = fmt.Sprintf("%d", 1)
	s := BuildBinMsgPack(&cache, cmdMap.URLFormEncode(), nil)
	return this.sendCmd(s, token)
}

func (this *WsBinMsgCli) innerUnSub(topic string) error {
	token := NewWsBinMsgToken()
	var cache gobase.BytesBuilder
	cmdMap := gobase.NewStrMap()
	cmdMap["token"] = token.token
	cmdMap["action"] = fmt.Sprintf("%d", 2)
	s := BuildBinMsgPack(&cache, cmdMap.URLFormEncode(), nil)
	return this.sendCmd(s, token)
}

func (this *WsBinMsgCli) resubAll(dna int32) {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}
	this.subLk.Lock()
	defer this.subLk.Unlock()
	for k, v := range this.subMap {
		if dna != atomic.LoadInt32(&this.connectDNA) {
			break // 重连了
		}
		if v != dna {
			if v == -1 {
				delete(this.subMap, k)
			} else {
				if this.innerSub(k) != nil {
					this.subMap[k] = dna
				}
			}
		}
	}
}

func (this *WsBinMsgCli) Sub(topic string) {
	this.subLk.Lock()
	defer this.subLk.Unlock()
	if _, ok := this.subMap[topic]; !ok {
		if this.innerSub(topic) == nil {
			this.subMap[topic] = this.connectDNA
		} else {
			this.subMap[topic] = 0
		}
	}
}

func (this *WsBinMsgCli) UnSub(topic string) {
	this.subLk.Lock()
	defer this.subLk.Unlock()
	delete(this.subMap, topic)
	this.innerUnSub(topic)
}

func (this *WsBinMsgCli) Pub(topic string, msg []byte) (n int, err error) {
	token := NewWsBinMsgToken()
	var cache gobase.BytesBuilder
	cmdMap := gobase.NewStrMap()
	cmdMap["action"] = fmt.Sprintf("%d", 0)
	cmdMap["topic"] = topic
	cmdMap["token"] = token.token
	s := BuildBinMsgPack(&cache, cmdMap.URLFormEncode(), msg)
	err = this.sendCmd(s, token)
	if err != nil {
		return -1, err
	} else {
		respMap := gobase.NewStrMap()
		respMap.URLFormDecode(string(token.respBuf))
		n = respMap.IntByName("n", 0)
		return n, nil
	}
}

func (this *WsBinMsgCli) sendCmd(binmsgPack []byte, token *WsBinmsgToken) (err error) {
	this.cmdLk.Lock()
	this.cacheCmd[token.token] = token
	this.cmdLk.Unlock()
	defer func() {
		this.cmdLk.Lock()
		delete(this.cacheCmd, token.token)
		this.cmdLk.Unlock()
	}()
	err = this.wscli.SendMsg(websocket.BinaryFrame, binmsgPack)
	if err != nil {
		return err
	}
	if this.WaitRespFlag == 1 {
		if token.Wait(this.respTimeOut) {
			return nil
		} else {
			return fmt.Errorf("wait resp timeout")
		}
	} else {
		return nil
	}
}
